// @Author EthanScriptOn
// @Desc
package op_example

import (
	"context"
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/custom_exp"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_basic_components/util"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
)

type TimeoutDetectionMultiChannelDataOperator struct {
	*identity.IdentityInformationCarrier
}

func (t *TimeoutDetectionMultiChannelDataOperator) Execute(ctx *actuator.SrfSwitchBridgeActuatorContext, configIdentifier interface{}, structureParam interface{}) (resp interface{}, err error) {
	var operatorConfig *operator.TimeOutMultiChannelDataOperatorConfig
	if operatorConfig, err = t.GetOperatorConfig(configIdentifier); err != nil {
		return nil, err
	}
	ctxWithTimeOut, _ := context.WithTimeout(context.Background(), operatorConfig.TimeOut)
	return t.ExecuteWithContext(ctxWithTimeOut, ctx, operatorConfig, structureParam)
}

func (t *TimeoutDetectionMultiChannelDataOperator) GetOperatorConfig(configIdentifier interface{}) (*operator.TimeOutMultiChannelDataOperatorConfig, error) {
	if configIdentifier == nil {
		return nil, fmt.Errorf("config identifier is nil")
	}
	config := operator.GetOperatorConfig(configIdentifier)
	if config == nil {
		return nil, fmt.Errorf(fmt.Sprintf("can not found extend_operator config with: %+v", configIdentifier))
	}
	configResp, ok := config.(*operator.TimeOutMultiChannelDataOperatorConfig)
	if !ok {
		return nil, fmt.Errorf(fmt.Sprintf("extend_operator config type mismatch: %+v", configIdentifier))
	}
	return configResp, nil
}

func (t *TimeoutDetectionMultiChannelDataOperator) ExecuteWithContext(ctxWithTimeOut context.Context, ctx *actuator.SrfSwitchBridgeActuatorContext, config *operator.TimeOutMultiChannelDataOperatorConfig, params interface{}) (interface{}, error) {
	var executeResponse interface{}
	var err error
	signal := make(chan struct{})
	go func() {
		defer util.HandleErr(&err)
		executeResponse, err = t.DoExecute(ctx, config, params)
		close(signal)
	}()
	select {
	case <-ctxWithTimeOut.Done():
		return nil, custom_exp.GenerateTimeOutError("TaskOperator call is already timeOut")
	case <-signal:
		return executeResponse, err
	}
}

func (t *TimeoutDetectionMultiChannelDataOperator) DoExecute(ctx *actuator.SrfSwitchBridgeActuatorContext, config *operator.TimeOutMultiChannelDataOperatorConfig, params interface{}) (resp interface{}, err error) {
	operatorIdentifier := config.GetOperatorIdentifier(true)
	if operatorIdentifier == nil {
		return nil, fmt.Errorf("can not found operatorIdentifier")
	}
	operatorInstance := operator.GetOperator(operatorIdentifier)
	if operatorInstance == nil {
		return nil, fmt.Errorf(fmt.Sprintf("can not found extend_operator with: %+v", operatorIdentifier))
	}
	return operatorInstance.Execute(ctx, config.GetOperatorConfigIdentifier(true), params)
}
